package com.example.datastreamdemo

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import java.util.concurrent.TimeUnit

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

object MySourceAichParlleFunctionDemo {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val function = new MyRichParalleSourceFunction()
    val res: DataStream[Student] = senv.addSource(function).setParallelism(1)
    res.print()


    senv.execute()
    // 可以不需要等待作业完成，executeAsync() 异步执行

  }
}
case class Student(id: Long, name: String,  age:Long)
class MyRichParalleSourceFunction extends RichSourceFunction[Student] {
  var ele:Long = 0
  var isRunning = true

  var connection: Connection = null
  var cp: PreparedStatement  = null

  override def cancel(): Unit = {
    isRunning = false
  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
    while(isRunning) {
      ele += 1;
      val res: ResultSet = cp.executeQuery()
      while (res.next()){
        val id = res.getInt("id")
        val name = res.getString("name")
        val age = res.getInt("age")
        val stu: Student = new Student(id, name, age)
        ctx.collect(stu)
      }

      TimeUnit.SECONDS.sleep(3)
    }
  }

  override def open(parameters: Configuration): Unit = {
    connection = DriverManager.getConnection("jdbc:mysql://localdb:3307/db_demo", "root", "123456")
    cp = connection.prepareStatement("select * from student")
  }

  override def close(): Unit = {
    if (cp != null){
      cp.close()
    }
    if (connection != null){
      connection.close()
    }
  }
}
